package com.cat.net.network.kafka.serializer;

import com.cat.net.network.base.Packet;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.kafka.common.serialization.Deserializer;

/**
 * 字节流转Packet反序列化方式
 * @author Jeremy
 */
public class KafkaPacketDeserializer implements Deserializer<Packet> {

    @Override
    public Packet deserialize(String s, byte[] bytes) {
        if (bytes == null)
            return null;

        ByteBuf buffer = Unpooled.wrappedBuffer(bytes);
        int cmd = buffer.readInt();
        int seq = buffer.readInt();
        byte[] payload = new byte[buffer.readableBytes()];
        buffer.readBytes(payload);
        buffer.release();
        return new Packet(cmd, seq, payload);
    }

}
